package com.pie4cloud.pie.job.executor.job;

import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.pie4cloud.pie.bus.api.consumer.MetricCalcService;
import com.pie4cloud.pie.bus.api.dto.MessageDto;
import com.pie4cloud.pie.common.core.constant.SecurityConstants;
import com.pie4cloud.pie.common.core.util.R;
import com.pie4cloud.pie.job.api.entity.*;
import com.pie4cloud.pie.job.api.commons.Constant;
import com.pie4cloud.pie.job.api.consumer.IHanlderConsumer;
import com.pie4cloud.pie.job.api.consumer.annotation.Hanlder;
import com.pie4cloud.pie.job.api.dto.CallBackDto;
import com.pie4cloud.pie.job.api.feign.RemoteDispenseCongfigService;
import com.pie4cloud.pie.job.api.utils.FtpUtils;
import com.pie4cloud.pie.job.api.vo.TimeLevelVo;
import com.pie4cloud.pie.job.executor.dto.StatisticalRequest;
import com.pie4cloud.pie.job.executor.enums.StatisticalType;
import com.pie4cloud.pie.job.executor.service.DispenseToolService;
import com.pie4cloud.pie.job.executor.services.ConfigDecodeService;
import com.pie4cloud.pie.job.executor.services.ConfigGatherService;
import com.pie4cloud.pie.job.executor.services.StatisticalService;
import com.pie4cloud.pie.job.executor.utils.*;
import com.pie4cloud.pie.job.executor.services.ConfigFtpService;
import com.pie4cloud.pie.job.executor.utils.ftp.FtpProcessor;
import com.pie4cloud.pie.job.executor.utils.ftp.FtpUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FilenameUtils;
import org.apache.commons.net.ftp.FTPClient;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.*;

@Slf4j
@Service
@Hanlder(handler = "gatherHandler")
public class GatherHandlerService implements IHanlderConsumer {

	@Resource
	private ConfigGatherService configGatherService;

	@Resource
	private ConfigFtpService configFtpService;

	@Resource
	private ConfigDecodeService configDecodeService;

	@Resource
	private MetricCalcService metricCalcService;

	@Resource
	private RedisTemplate<String,String> redisTemplate;
	@Resource
	private RemoteDispenseCongfigService remoteDispenseCongfigService;
	@Resource
	private DispenseToolService dispenseToolService;
	@Resource
	private StatisticalService statisticalService;
	@Resource
	private FtpUtil ftpUtil;

	private static final String redis_key_start = "gather";

//	@Value(value = "#{'gather.file.local.cache'}")
//	private String cacheFilePath;


	@Override
	public R execute(String jobInfoEntity, CallBackDto callBackDto) throws Exception {

		GatherEntity gatherEntity = JSON.parseObject(jobInfoEntity, GatherEntity.class);
		Long gatherId = gatherEntity.getGatherId();
		Long decodeId = gatherEntity.getDecodeId();
		ConfigGather configGather = configGatherService.getById(gatherId);
		ConfigDecode configDecode = configDecodeService.getById(decodeId);
		DecodeParams decodeParams = JSON.parseObject(configDecode.getDecodeParams(), DecodeParams.class);
		TimeLevelVo timeLevelVo=new TimeLevelVo();
		R<PlaybillEntity> playbillEntityR=remoteDispenseCongfigService.findByDdataId(configDecode.getPlayBillId(), SecurityConstants.FROM_IN);
		PlaybillEntity playbillEntity=playbillEntityR.getData();
		//测试统计-开始
		SimpleDateFormat format=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

		Date lastTime=dispenseToolService.getLastTriggerTime(playbillEntity.getJobCron(),format.parse("2021-05-07 01:00:00"));
		Date nextTime=dispenseToolService.getNextTriggerTime(playbillEntity.getJobCron(),lastTime);
		timeLevelVo.setStartTime(lastTime.getTime());
		timeLevelVo.setEndTime(nextTime.getTime());
		timeLevelVo.setTimeKey(lastTime.getTime());

		//测试统计-结束
		if(Constant.GATHER_SCAN_TYPE_FTP.equals(configGather.getGatherScanType())) {
			ConfigFtp configFtp = configFtpService.getById(configGather.getGatherFtpId());

				DirectoryUtils directoryUtils = new FtpDirectoryUtils();
				List<String> allBaseDirectory = directoryUtils.getAllBaseDirectory(configGather.getGatherBaseAddress(), new Date(gatherEntity.getTriggerLastTime()));
				Set<GatherMessageEntity> allFile = new HashSet<>();
				String pathReg = decodeParams.getPathReg();
				List<String> ftpPath=new ArrayList<>();
				FtpProcessor ftpProcessor=ftpUtil.getFtpProcessor(configFtp);
			    allBaseDirectory.forEach(dire->{
					ftpProcessor.recursiveFileNames(dire,ftpPath);
				});
				ftpPath.forEach(fpath->{
					GatherMessageEntity gatherMessageEntity=new GatherMessageEntity();
					String[] match = new String[0];
					try {
						match = com.pie4cloud.pie.job.api.utils.RegUtils.match(pathReg, fpath);
						gatherMessageEntity.setRemoteFilePath(fpath);
						gatherMessageEntity.setRegList(Arrays.asList(match));
						allFile.add(gatherMessageEntity);
					} catch (com.pie4cloud.pie.job.api.utils.RegException e) {
						log.warn("file {} is not match {}",fpath,pathReg);
					}

				});
				//FtpUtils.scanAllFile(ftpClient,allBaseDirectory,pathReg,allFile);
				for (GatherMessageEntity gatherMessageEntity: allFile) {

					HashMap<String, String> commonMap = new HashMap<>();
					commonMap.put("fileName",FilenameUtils.getName(gatherMessageEntity.getRemoteFilePath()));
					Map<String, String> decode = DecodeUtils.decode(gatherMessageEntity.getRemoteFilePath(),
							pathReg, DecodeUtils.getParamsMap(decodeParams.getDecodeConfig()),commonMap,"localFilePath");
					Map<String, String> map = RegUtils.matchToMap(pathReg, gatherMessageEntity.getRemoteFilePath());
					String parse = PathUtils.parse(configGather.getLocalBaseAddress(), decode, map);
					//if (!redisTemplate.opsForSet().isMember(redis_key_start+gatherEntity.getId(),gatherMessageEntity.getRemoteFilePath())) {
						String name = FilenameUtils.getName(parse);
						String path = FilenameUtils.getFullPath(parse);
						StatisticalRequest statisticalRequest=new StatisticalRequest();
						statisticalRequest.setStatisticalType(StatisticalType.GATHER);
						statisticalRequest.setPlaybillId(configDecode.getPlayBillId());
						statisticalRequest.setFilePath(path+"/"+name);
						statisticalRequest.setWhenTime(timeLevelVo.getTimeKey());
						if(!statisticalService.check(statisticalRequest)){
							ftpProcessor.downloadFile(gatherMessageEntity.getRemoteFilePath(),name,path);
							//FtpUtils.download(ftpClient,gatherMessageEntity.getRemoteFilePath(),path, name);
							//redisTemplate.opsForSet().add(redis_key_start+gatherEntity.getId(),parse);
							gatherMessageEntity.setLocalFilepath(parse);
							commonMap.put("localFilePath",parse);
							PathUtils.dealIgnoreKeys(decode,commonMap);
							gatherMessageEntity.setDecodeResult(decode);
							gatherMessageEntity.setTimeLevel(timeLevelVo.getTimeKey());
							gatherMessageEntity.setTableName(decodeParams.getTableName());
							gatherMessageEntity.setPlayBillId(configDecode.getPlayBillId());
							ObjectMapper objectMapper = new ObjectMapper();
							System.out.println(objectMapper.writeValueAsString(gatherMessageEntity));
							this.excuteKaf(gatherMessageEntity,"GATHER"+"_"+configDecode.getPlayBillId());
							statisticalService.doOperation(statisticalRequest);
						}

					}

				//FtpUtils.disconnect(ftpClient);

		} else if(Constant.GATHER_SCAN_TYPE_LOCAL.equals(configGather.getGatherScanType())) {
		}else {

		}
		callBackDto.setParams(jobInfoEntity);
		return R.ok();
	}

	public void excuteKaf(GatherMessageEntity gatherMessageEntity,String topic){
		List<MessageDto> messageDtos=new ArrayList<>();
		MessageDto messageDto=new MessageDto();
		messageDto.setContent(JSON.toJSONString(gatherMessageEntity));
		messageDto.setTopic(topic);
		messageDtos.add(messageDto);
		metricCalcService.submit(messageDtos);
	}
}
